package com.zr.flink;

import com.example.klinemqflink.model.MarketData;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.json.JsonMapper;
import com.zr.flink.thread.DataProcessor;
import com.zr.flink.thread.ThreadDemo;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;


//@Service
public class MyMessageListener {
    private static final Map<String, ExecutorService> threadMap = new HashMap<>();

    @Autowired
    private DataProcessor dataProcessor;

    @RabbitListener(queues = {"market.data.queue.1_1", "market.data.queue.1_2", "market.data.queue.2_1","market.data.queue.2_2"})
    public void receiveMessage(String message) throws InterruptedException, JsonProcessingException {
//        MarketData marketData = new JsonMapper().readValue(message, MarketData.class);
//        ExecutorService executorService = null;
//        String threadName = marketData.getChannel();

        System.out.println(message);
//        if (threadMap.containsKey(threadName)) {
//            executorService = threadMap.get(threadName);
//        } else {
//            executorService = Executors.newSingleThreadExecutor();
//            threadMap.put(threadName, executorService);
//        }
//        dataProcessor.setThreadName(threadName);
//        dataProcessor.setMarketData(marketData);
//        executorService.submit(new ThreadDemo(dataProcessor));
    }
}